package com.zidongxiangxi.rabbit.consumer.job;

import com.zidongxiangxi.rabbit.consumer.config.ConsumeRecordClearConfig;
import com.zidongxiangxi.rabbit.consumer.manager.ConsumeRecordManager;
import com.zidongxiangxi.rabbit.core.job.AbstractWorkThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;

/**
 * 消费记录清理任务
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class ConsumeRecordClearThread extends AbstractWorkThread {
	private final Logger logger = LoggerFactory.getLogger(ConsumeRecordClearThread.class);

	private final long retentionPeriod;
	private final long workPeriod;
	private final int batchSize;
	private final ConsumeRecordManager consumeRecordManager;

	public ConsumeRecordClearThread(
			ConsumeRecordClearConfig config,
			ConsumeRecordManager consumeRecordManager
	) {
		this.retentionPeriod = config.getRetentionPeriod() * 86400 * 1000;
		this.workPeriod = config.getWorkPeriod();
		this.batchSize = config.getClearBatchSize();
		this.consumeRecordManager = consumeRecordManager;
	}

	@Override
	public void work() {
		Date beforeTime = new Date(System.currentTimeMillis() - retentionPeriod);
		int size = clearRecord(beforeTime, batchSize);
		if (size < batchSize) {
			try {
				Thread.sleep(workPeriod);
			} catch (InterruptedException ignore) {}
		}
	}

	private int clearRecord(Date beforeTime, int batchSize) {
		try {
			List<Long> ids = consumeRecordManager.listIds(beforeTime, batchSize);
			if (CollectionUtils.isEmpty(ids)) {
				return 0;
			}
			consumeRecordManager.deleteRecords(ids);
			return ids.size();
		} catch (Throwable e) {
			logger.error("清理消费记录失败", e);
		}
		return 0;
	}
}
